/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package users;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;

import org.apache.qpid.AMQException;
import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.client.message.JMSTextMessage;
import org.apache.qpid.url.URLSyntaxException;

/**
 * Data needed by a publisher to connect with a Qpid broker and interact
 * with it.
 */
public class PublisherConnection extends UserConnection {

    /** List of topics to publish to. */
    private Set<String> topics = new HashSet<String>();
    /** Mapping of topics to publish to and publishers for them. */
    private HashMap<String, TopicPublisherAdapter> topicPub
            = new HashMap<String, TopicPublisherAdapter>();

    /**
     * Getter for the variable 'topics'.
     * @return variable 'topics'
     */
    public Set<String> getTopics() {
        return topics;
    }
    /**
     * Getter for the variable 'topicPub'.
     * @return variable 'topicPub'
     */
    public HashMap<String, TopicPublisherAdapter> getTopicPub() {
        return topicPub;
    }


    /**
     * Constructor.
     *
     * @param publisher  to whom the connection belongs
     * @param broker  to connect to
     * @param port  in which the broker is listening
     * @param exchange  to use in the broker
     */
    public PublisherConnection(final Publisher publisher, final String broker,
                               final int port, final String exchange) {
        super(publisher, broker, port, exchange);
    }

    /**
     * Constructor using the default port.
     *
     * @param publisher  to whom the connection belongs
     * @param broker  to connect to
     * @param exchange  to use in the broker
     */
    public PublisherConnection(final Publisher publisher, final String broker,
                               final String exchange) {
        super(publisher, broker, exchange);
    }

    /**
     * Get the connection ready.
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    @Override
    public void initialize()
            throws JMSException, URLSyntaxException, AMQException {
        super.initialize();
        for (String topic : getTopics()) {
            initializeTopic(topic);
        }
    }

    /**
     * Create a publisher for a topic and add it to the hashmap of publishers.
     *
     * @param topicName  name of the topic to publish to
     * @throws JMSException  if there is any problem
     */
    private void initializeTopic(final String topicName) throws JMSException {
        //delete the previous publisher, if there was one
        getTopicPub().remove(topicName);

        //get publisher for broker
        TopicPublisherAdapter publisher = getPublisherForTopic(topicName);
        //put topic and publisher in hashmap
        getTopicPub().put(topicName, publisher);

//        System.out.println("Publisher to topic \""
//                           + publisher.getTopic() + "\" created");
    }

    /**
     * Add a topic to the list of topics to publish about, create a publisher
     * for it in the brokers and add it to the hashmap of topics and
     * publishers.
     *
     * @param topicName  the name of the topic to be added
     * @throws JMSException  if there is any problem
     */
    public void addTopic(final String topicName) throws JMSException {
        //add topic to list of topics
        getTopics().add(topicName);
        //initialize it
        initializeTopic(topicName);
    }

    /**
     * Remove a topic from the list of topics to publish about and close its
     * publisher.
     *
     * @param topicName  the name of the topic to be removed
     * @throws JMSException  if there is any problem
     */
    public void removeTopic(final String topicName) throws JMSException {
        //remove topic from list of topics
        getTopics().remove(topicName);
        //close publisher
        getTopicPub().get(topicName).close();
    }

    /**
     * Publish a text message to the broker.
     *
     * @param topicName  topic to send the message to
     * @param text  text of the message
     * @param properties  properties of the message
     * @param correlationID  correlation ID of the message
     * @throws JMSException 
     */
    public void publishMessage(final String topicName, final String text,
            final HashMap<String, String> properties,
            final String correlationID) throws JMSException {

        //get the publisher in the broker
        TopicPublisherAdapter publisher = getTopicPub().get(topicName);
        //create the message to send
        JMSTextMessage txtMsg
                = (JMSTextMessage) getSession().createTextMessage(text);
        //set the properties of the message
        if (properties != null) {
            Set<Map.Entry<String, String>> propEntries = properties.entrySet();
            for (Map.Entry<String, String> entry : propEntries) {
                txtMsg.setStringProperty(entry.getKey(), entry.getValue());
            }
        }
        //set the correlation ID of the message
        txtMsg.setJMSCorrelationID(correlationID);
        //send the message created
        publisher.publish(txtMsg, DeliveryMode.PERSISTENT,
                          AbstractJMSMessage.DEFAULT_PRIORITY,
                          AbstractJMSMessage.DEFAULT_TIME_TO_LIVE);
        //print info about the message sent
//            System.out.println("Published message: \"" + txtMsg.getText()
//                             + "\" to " + txtMsg.getJMSDestination() + " in "
//                             + getBROKER() + " at port " + getPORT());
        System.out.println("Published message: \"" + txtMsg.getText()
                           + "\" to " + topicName + " in "
                           + getBroker() + " at port " + getPort());

    }

}
